package com.interview.lock;

import java.util.List;
import java.util.concurrent.atomic.AtomicBoolean;

/**
 * 消费者
 * 
 * 设计要点：
 * 1. 单个消费者，不需要考虑多消费者竞争
 * 2. 使用批量消费提高效率
 * 3. 支持优雅停止
 */
public class Consumer implements Runnable {
    
    // 消息队列
    private final MessageQueue messageQueue;
    // 批量消费大小
    private final int batchSize;
    // 是否继续运行
    private final AtomicBoolean running;
    // 期望消费的总消息数（用于自动停止）
    private final long expectedMessageCount;
    
    public Consumer(MessageQueue messageQueue, int batchSize, long expectedMessageCount) {
        this.messageQueue = messageQueue;
        this.batchSize = batchSize;
        this.running = new AtomicBoolean(true);
        this.expectedMessageCount = expectedMessageCount;
    }
    
    @Override
    public void run() {
        System.out.println("消费者启动，批量大小: " + batchSize);
        
        long processedCount = 0;
        long startTime = System.currentTimeMillis();
        
        try {
            while (running.get()) {
                // 批量获取消息
                List<Message> messages = messageQueue.takeBatch(batchSize);
                
                if (messages != null && !messages.isEmpty()) {
                    // 处理消息
                    processMessages(messages);
                    processedCount += messages.size();
                    
                    // 定期打印进度
                    if (processedCount % 100000 == 0) {
                        long elapsed = System.currentTimeMillis() - startTime;
                        double rate = processedCount * 1000.0 / elapsed;
                        System.out.printf("已处理 %d 条消息，速率: %.2f 条/秒%n", 
                            processedCount, rate);
                    }
                    
                    // 如果达到期望的消息数，停止消费
                    if (expectedMessageCount > 0 && processedCount >= expectedMessageCount) {
                        System.out.println("已达到期望消息数: " + expectedMessageCount);
                        break;
                    }
                }
            }
            
        } catch (InterruptedException e) {
            System.out.println("消费者被中断");
            Thread.currentThread().interrupt();
        } catch (Exception e) {
            System.err.println("消费者发生错误: " + e.getMessage());
            e.printStackTrace();
        } finally {
            long endTime = System.currentTimeMillis();
            long totalTime = endTime - startTime;
            double avgRate = processedCount * 1000.0 / totalTime;
            
            System.out.println("\n=== 消费者统计信息 ===");
            System.out.println("总处理消息数: " + processedCount);
            System.out.println("总耗时: " + totalTime + " 毫秒");
            System.out.printf("平均处理速率: %.2f 条/秒%n", avgRate);
            System.out.println("====================\n");
        }
    }
    
    /**
     * 处理消息
     * 
     * @param messages 消息列表
     */
    private void processMessages(List<Message> messages) {
        // 这里可以实现具体的业务逻辑
        // 为了演示，我们只是简单地"处理"这些消息
        
        for (Message message : messages) {
            // 模拟消息处理
            // 在实际应用中，这里可能是：
            // - 写入数据库
            // - 发送到其他系统
            // - 执行业务逻辑
            // 等等
            
            // 可选：打印部分消息用于调试
            // if (message.getId() % 10000 == 0) {
            //     System.out.println("处理消息: " + message);
            // }
        }
    }
    
    /**
     * 停止消费者
     */
    public void stop() {
        running.set(false);
        System.out.println("消费者停止信号已发送");
    }
    
    /**
     * 检查是否还在运行
     * 
     * @return 是否运行中
     */
    public boolean isRunning() {
        return running.get();
    }
}

